package com.xyy.saas.payment.util.mq;

import cn.hutool.core.thread.ThreadUtil;
import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ctrip.framework.apollo.Config;
import com.ctrip.framework.apollo.ConfigChangeListener;
import com.ctrip.framework.apollo.ConfigService;
import com.ctrip.framework.apollo.model.ConfigChange;
import com.ctrip.framework.apollo.model.ConfigChangeEvent;
import com.google.common.util.concurrent.RateLimiter;
import com.xyy.ec.order.api.query.OrderApi;
import com.xyy.ec.order.dto.pay.OrderTransDto;
import com.xyy.saas.payment.adpater.impl.JdCreditAdapter;
import com.xyy.saas.payment.adpater.jdCredit.resp.LoanInfoResponse;
import com.xyy.saas.payment.adpater.pingxx.domain.PayDO;
import com.xyy.saas.payment.callback.core.dto.CallbackMsg;
import com.xyy.saas.payment.callback.core.result.CallbackRunner;
import com.xyy.saas.payment.common.Constants;
import com.xyy.saas.payment.common.RocketMqGroup;
import com.xyy.saas.payment.common.RocketMqTag;
import com.xyy.saas.payment.common.RocketMqTopic;
import com.xyy.saas.payment.common.model.Tuple2;
import com.xyy.saas.payment.common.pool.NamedTask;
import com.xyy.saas.payment.common.pool.PoolUtil;
import com.xyy.saas.payment.converter.PayCommonConverter;
import com.xyy.saas.payment.cores.enums.BusinessIdTypeEnum;
import com.xyy.saas.payment.cores.enums.ChannelEnum;
import com.xyy.saas.payment.cores.vo.CallbackResultVo;
import com.xyy.saas.payment.dao.mapper.PayResultOrderMapper;
import com.xyy.saas.payment.dao.model.LoanRepayPlanDetailPo;
import com.xyy.saas.payment.dao.model.LoanRepayPlanPo;
import com.xyy.saas.payment.dao.model.OrderTransDataPo;
import com.xyy.saas.payment.dao.model.PayResultOrder;
import com.xyy.saas.payment.manager.PayManager;
import com.xyy.saas.payment.merchant.core.dto.RechargeHandleDto;
import com.xyy.saas.payment.service.PayResultOrderService;
import com.xyy.saas.payment.trace.AlarmService;
import com.xyy.saas.payment.util.*;
import org.apache.commons.lang.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;

/**
 * @author wangtianqi
 * @date 2020/2/28 16:56
 */
@DependsOn("apolloUtil")
@Service
@RocketMQMessageListener(topic = RocketMqTopic.TOPIC_PINGXX_CALLBACK, consumerGroup = RocketMqGroup.GROUP_PINGXX_CALLBACK, consumeMode = ConsumeMode.CONCURRENTLY)
public class  PingXXCallback  implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener, InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(WxCallback.class);
    @Autowired
    private CallbackRunner callbackRunner;
    @Autowired
    private PayResultOrderService payResultOrderService;
    @Autowired
    private MqProducer mqProducer;
    @Resource
    private SkyWalkingMonitorUtil skyWalkingMonitorUtil;
    private int rateLimiterCreate = Integer.parseInt(ApolloUtil.getProperty(Constants.WX_CARLLBACK_RATE_LIMITER_CREATE));
    private RateLimiter rateLimiter = RateLimiter.create(rateLimiterCreate);

    @Autowired
    private RedisUtil redisUtil;

    @Autowired
    private PayResultOrderMapper payResultOrderMapper;
    @Autowired
    private PayManager payManager;
    @Reference(version = "1.0.0")
    private OrderApi orderApi;
    @Resource
    private JdCreditAdapter jdCreditAdapter;
    @Override
    public void onMessage(MessageExt messageExt) {
        rateLimiter.acquire(1);
        String body = new String(messageExt.getBody());
        PayDO payDO = JSONUtils.parse2Object(body,  PayDO.class);

        if(payDO==null) {
            LOGGER.error("PingXXCallback callBackHandler pingxxPay callback fail domain is null");
            return;
        }
        String lockKey = RedisUtil.getPayResultPrefix(payDO.getOrderNo());
        if (!redisUtil.tryLock(lockKey,redisUtil.SECOND_5)) {
            LOGGER.warn("PingXXCallback callback get lock fail {}",payDO.getOrderNo());
            throw new RuntimeException();
        }
        try {
            String tags = messageExt.getTags();
            if (Objects.equals(RocketMqTag.TOPIC_PINGXX_CALLBACK_QUERY_TRADE_TAG, tags)) {
                handleTransactionPayQueryCallBack(payDO);
                return;
            }

            PayResultOrder payResultOrder = payResultOrderService.selectByPayNoAndChannel(payDO.getOrderNo(), payDO.getChannel());

            if(payResultOrder != null && Objects.equals(payResultOrder.getPayChannel(), payDO.getChannel())) {
                LOGGER.error("PingXXCallback callBackHandler pingxxPay callback repeat payNo:{}", payDO.getOrderNo());
                return;
            }

            payResultOrder =  buildPayResultOrder(payDO);

            int count = payResultOrderMapper.insertSelective(payResultOrder);;

            if(count<1){
                LOGGER.error("PingXXCallback callBackHandler pingxxPay callback insert payResultOrder fail payResultOrder:{}", JSON.toJSON(payResultOrder));
                throw new Exception();
            }

            // 异步处理，不阻塞主流程
            afterPaySuccess(payDO, payResultOrder);

            Long unixTime = System.currentTimeMillis()/1000;
            String channel = null;

            // 若为京东银行卡支付，则传递原channel
            if (Constants.thirdChannelSubmitPingan(payDO.getChannel()) || BusinessIdTypeEnum.EC.getType().equals(payDO.getBusinessType())) {
                channel = payDO.getChannel();
            } else if (ChannelEnum.JD_CREDIT.getCode().equals(payDO.getChannel())) {
                channel = payDO.getChannel();
            } else {
                if (payDO.getChannel().contains("_")) {
                    channel = Constants.CHANNEL_AGGREGATE_NAME + "_" + payDO.getChannel().split("_")[1];
                } else {
                    channel = Constants.CHANNEL_AGGREGATE_NAME + "_" + payDO.getChannel();
                }
            }
            // 异步回调业务
            String topic = getPaySuccessNotifyTopic(payDO.getBusinessType(), payDO.getBusinessOrderType());
            CallbackMsg msg = CallbackMsg.build(payResultOrder.getPayNo(), CallbackResultVo.TYPE_CHARGE_SUCCESS,
                    payDO.getAmount(), payResultOrder.getFee(), unixTime, payDO.getCallbackAddress(), channel, DateUtil.now(), payDO.getMchId());
            msg.setBusinessOrderNo(payDO.getBusinessOrderNo());
            msg.setChannelOrderNo(payResultOrder.getChannelTransactionNo());
            callbackRunner.callback(topic, msg);

            // 异步处理渠道充值
            asyncRechargeHandle(payDO, payResultOrder);

            LOGGER.info("PingXXCallback callBackHandler pingxxPay end payResultOrder:{}, topic:{}, msg:{}", JSON.toJSON(payResultOrder), topic, JSONObject.toJSONString(msg));

            // 支付回调，支付平台保存订单耗时
            skyWalkingMonitorUtil.summary(SkyWalkingMonitorUtil.CALLBACK_PAYMENT_SAVE_COST, DateUtil.getTimeDiff(payDO.getNotifyTime(), DateUtil.now()));

//            profitSharingService.jdCreditPaySettle(payResultOrder.getPayNo());
        }
        catch(Exception e) {
            LOGGER.error("PingXXCallback callBackHandler error domain:{}", JSON.toJSON(payDO), e);

            if(payDO.getCount() == 16) {
                LOGGER.error("PingXXCallback callBackHandler error is 16 times domain:{}", JSON.toJSON(payDO));
                return ;
            }

            payDO.setCount(payDO.getCount() + 1);
            mqProducer.syncSendDelayMsg(RocketMqTopic.TOPIC_PINGXX_CALLBACK, JSON.toJSONString(payDO), 1);
        } finally {
            redisUtil.unLock(lockKey);
        }
    }

    private void handleTransactionPayQueryCallBack(PayDO payDO) {
        LOGGER.info("PingXXCallback handleTransactionPayQueryCallBack payDO:{}", JSON.toJSONString(payDO));
        PayResultOrder payResultOrder = payResultOrderService.selectByPayNoAndChannel(payDO.getOrderNo(), payDO.getChannel());
        if (payResultOrder == null) {
            LOGGER.error("PingXXCallback handleTransactionPayQueryCallBack payResultOrder is null payNo:{}", payDO.getOrderNo());
            return;
        }

        Long unixTime = System.currentTimeMillis() / 1000;
        String channel;

        // 若为京东银行卡支付，则传递原channel
        if (Constants.thirdChannelSubmitPingan(payDO.getChannel()) || BusinessIdTypeEnum.EC.getType().equals(payDO.getBusinessType())) {
            channel = payDO.getChannel();
        } else if (ChannelEnum.JD_CREDIT.getCode().equals(payDO.getChannel())) {
            channel = payDO.getChannel();
        } else {
            if (payDO.getChannel().contains("_")) {
                channel = Constants.CHANNEL_AGGREGATE_NAME + "_" + payDO.getChannel().split("_")[1];
            } else {
                channel = Constants.CHANNEL_AGGREGATE_NAME + "_" + payDO.getChannel();
            }
        }
        // 异步回调业务
        String topic = getPaySuccessNotifyTopic(payDO.getBusinessType(), payDO.getBusinessOrderType());
        CallbackMsg msg = CallbackMsg.build(payResultOrder.getPayNo(), CallbackResultVo.TYPE_CHARGE_SUCCESS,
                payDO.getAmount(), payResultOrder.getFee(), unixTime, payDO.getCallbackAddress(), channel, DateUtil.now(), payDO.getMchId());
        msg.setBusinessOrderNo(payDO.getBusinessOrderNo());
        msg.setChannelOrderNo(payResultOrder.getChannelTransactionNo());
        callbackRunner.callback(topic, msg);

        // 异步处理渠道充值
        asyncRechargeHandle(payDO, payResultOrder);

        LOGGER.info("PingXXCallback handleTransactionPayQueryCallBack end payResultOrder:{}, topic:{}, msg:{}", JSON.toJSON(payResultOrder), topic, JSONObject.toJSONString(msg));
    }

    private void asyncRechargeHandle(PayDO payDO, PayResultOrder payResultOrder) {
        if (payDO.getChannel().contains(Constants.CHANNEL_DIRECT_NAME) || Constants.thirdChannelSubmitPingan(payDO.getChannel())) {
            RechargeHandleDto rechargeHandleDto = RechargeHandleDto.builder().
                    rechargeType(RechargeHandleDto.PAY).
                    channel(payResultOrder.getPayChannel()).
                    orderNo(payResultOrder.getPayNo()).
                    accountId(payResultOrder.getReceiveId()).
                    build();
            callbackRunner.syncSendMsg(RocketMqTopic.TOPIC_RECHARGE_HANDLE, rechargeHandleDto);
        }

        // 小雨点金融支付结算
        if (payDO.getChannel().contains(Constants.CHANNEL_XYD_LOAN)) {
            LOGGER.info("###小雨点支付回调发送支付结算mq, {}", payResultOrder.getPayNo());
            RechargeHandleDto rechargeHandleDto = RechargeHandleDto.builder().
                    rechargeType(RechargeHandleDto.PAY).
                    channel(payResultOrder.getPayChannel()).
                    orderNo(payResultOrder.getPayNo()).
                    accountId(payResultOrder.getReceiveId()).
                    build();
            callbackRunner.syncSendMsgWithTag(RocketMqTopic.TOPIC_RECHARGE_HANDLE, RocketMqTag.TOPIC_RECHARGE_HANDLE_XYD_CREDIT_PAY_SETTLE_TAG, rechargeHandleDto);
        }

    }

    private void afterPaySuccess(PayDO payDO, PayResultOrder payResultOrder) {
        // 农行e贷渠道，新增支付成功交易记录
        if (ChannelEnum.ABCHINA_LOAN.getCode().equals(payDO.getChannel()) || ChannelEnum.JD_CREDIT.getCode().equals(payDO.getChannel())) {
            // 异步查询京东融资渠道的还款计划
            PoolUtil.ALARM_POOL.execute(NamedTask.createRun(() -> {
                try {
                    OrderTransDto orderTransDto = orderApi.queryOrderTransData(payResultOrder.getPayNo()).getData();
                    // 现金实付取amount字段，此时查订单cashPayAmount字段后续可能会修改
                    OrderTransDataPo orderTransDataPo = PayCommonConverter.buildOrderTransDataPo(payDO, payResultOrder, orderTransDto);
                    LoanRepayPlanPo repayPlanPo = null;
                    List<LoanRepayPlanDetailPo> repayPlanDetailPos = null;
                    if (ChannelEnum.JD_CREDIT.getCode().equals(payDO.getChannel())) {
                        LoanInfoResponse response = retryQueryLoanDetail(payDO.getOrderNo());
                        Tuple2<LoanRepayPlanPo, List<LoanRepayPlanDetailPo>> tuple2 = PayCommonConverter.convertToRepayPo(response, orderTransDto.getOrderNo());
                        repayPlanPo = tuple2.getT1();
                        repayPlanDetailPos = tuple2.getT2();
                    }
                    payManager.insertOrderTrans(orderTransDataPo, repayPlanPo, repayPlanDetailPos);
                } catch (Exception e) {
                    LOGGER.error("PingXXCallback#repayPlan error:{}", JSON.toJSONString(payDO), e);
                    SpringUtils.getBean(AlarmService.class).asyncAlarm("支付成功新增交易数据失败", JSON.toJSONString(payDO));
                }
            }, "jd还款计划任务" + payResultOrder.getPayNo()));
        }
    }

    private LoanInfoResponse retryQueryLoanDetail(String orderNo) {
        LoanInfoResponse response = null;
        for (int i = 1; i <= 3; i++) {
            try {
                response = jdCreditAdapter.queryLoanDetail(null, orderNo);
                break;
            } catch (Exception e) {
                if (i == 3) {
                    throw e;
                }
                ThreadUtil.safeSleep(i * 200);
            }
        }
        return response;
    }


    private PayResultOrder buildPayResultOrder(PayDO payDO) {
        PayResultOrder  payResultOrder = new PayResultOrder();
        payResultOrder.setPayNo(payDO.getOrderNo());
        payResultOrder.setChannelChargeNo(payDO.getChannelChargeNo());
        payResultOrder.setPayChannel(payDO.getChannel());
        payResultOrder.setBusinessType(payDO.getBusinessType());
        payResultOrder.setBusinessOrderType(payDO.getBusinessOrderType());
        payResultOrder.setAmount(payDO.getAmount());
        //目前手续费业务方承担
        payResultOrder.setFee(0);
        payResultOrder.setReceiveId(payDO.getReceiverId());
        payResultOrder.setCtime(new Date());
        payResultOrder.setUtime(new Date());
        payResultOrder.setRefund((byte)0);
        payResultOrder.setRoyalty((byte)0);
        payResultOrder.setPayer(payDO.getPayer());
        payResultOrder.setPayMode(payDO.getPayMode());
        payResultOrder.setChannelTransactionNo(payDO.getChannelTransactionNo());
        payResultOrder.setMerchantId(payDO.getMchId());
        return payResultOrder;
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        defaultMQPushConsumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Set set = new HashSet();
        set.add(Constants.WX_CARLLBACK_RATE_LIMITER_CREATE);
        //动态监听配置变化
        Config config = ConfigService.getAppConfig();
        config.addChangeListener(new ConfigChangeListener() {
            @Override
            public void onChange(ConfigChangeEvent changeEvent) {
                ConfigChange change = changeEvent.getChange(Constants.WX_CARLLBACK_RATE_LIMITER_CREATE);
                rateLimiter = RateLimiter.create(Integer.parseInt(change.getNewValue()));
            }
        }, set);
    }

    /**
     * 业务线标识获取topic
     * @param businessType      业务线
     * @param businessOrderType 产品线
     * @return
     */
    private String getPaySuccessNotifyTopic(String businessType, String businessOrderType) {
        try {
            // 1.根据产品线获取topic
            String topic = ApolloUtil.getProperty("paySuccess_notify_topic_" + businessOrderType);

            if(!StringUtils.isBlank(topic)) {
                return topic;
            }

            // 2.根据业务线获取topic
            topic = ApolloUtil.getProperty("paySuccess_notify_topic_" + businessType);

            if(!StringUtils.isBlank(topic)) {
                return topic;
            }
        }
        catch(Exception ex) {
            LOGGER.error(ex.getMessage(), ex);
        }

        return RocketMqTopic.TOPIC_CALLBACK;
    }
}
